package com.niit;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.Random;
import org.apache.commons.lang3.RandomStringUtils;

public class demo {
    public static void main(String[] args) {
        // Kafka配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.136.128:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 创建Kafka生产者
        Producer<String, String> producer = new KafkaProducer<>(props);

        try {
            while (true) {
                // 生成随机数据
                String classId = RandomStringUtils.randomNumeric(8);
                String name = RandomStringUtils.randomAlphabetic(5);
                int gender = new Random().nextInt(2);
                String birth = "20211228"; // 示例日期，可修改
                String studentId = RandomStringUtils.randomNumeric(8);
                int semester = new Random().nextInt(8) + 1;
                char status = (new Random().nextInt(2) == 0)? 'L' : 'A';

                // 拼接数据
                String data = classId + "\t" + name + "\t" + gender + "\t" + birth + "\t" + studentId + "\t" + semester + "\t" + status;

                // 发送数据到Kafka主题
                producer.send(new ProducerRecord<>("stuInfo", data));
                System.out.println("Sent: " + data);

                // 每隔5秒发送一次数据
                Thread.sleep(5000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}